package com.abyss.job;


import com.abyss.avro.AvroDeserializeSchema;
import com.abyss.avro.SseAvro;
import com.abyss.avro.SzseAvro;
import com.abyss.bean.CleanBean;
import com.abyss.config.QuotConfig;
import com.abyss.map.SseMap;
import com.abyss.map.SzseMap;
import com.abyss.task.StockIncrTask;
import com.abyss.task.StockMinPutHdfsTask;
import com.abyss.task.StockMinTask;
import com.abyss.task.StockSecTask;
import com.abyss.util.QuotUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * @Date 2020/9/15
 * @Description 个股行情：秒级、分时、数据备份、涨跌幅、K线
 * 写入Druid之前，一定要先创建topic和开启Druid摄取数据的进程
 */
//1.创建StockStream单例对象，创建main方法
public class StockStream {

    public static void main(String[] args) throws Exception {
        /**
         * 个股总体开发步骤：
         * 1.创建StockStream单例对象，创建main方法
         * 2.获取流处理执行环境
         * 3.设置事件时间、并行度
         * 4.设置检查点机制
         * 5.设置重启机制
         * 6.整合Kafka(新建反序列化类)
         * 7.数据过滤（时间和null字段）
         * 8.数据转换、合并
         * 9.过滤个股数据
         * 10.设置水位线
         * 11.业务数据处理
         * 12.触发执行
         */
        //2.获取流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //3.设置事件时间、并行度
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//        //只在开发环境设置为1
        env.setParallelism(1);
//        //4.设置检查点机制
//        env.enableCheckpointing(5000);
//        env.setStateBackend(new FsStateBackend(QuotConfig.config.getProperty("stock.sec.hdfs.path")));
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        env.getCheckpointConfig().setFailOnCheckpointingErrors(false);
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //5.设置重启机制
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(5)));

        //6.整合Kafka(新建反序列化类)
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", QuotConfig.config.getProperty("bootstrap.servers"));
        properties.setProperty("group.id",QuotConfig.config.getProperty("group.id"));
        //消费沪市行情
        FlinkKafkaConsumer011<SseAvro> sseKafka = new FlinkKafkaConsumer011<SseAvro>(QuotConfig.config.getProperty("sse.topic"), new AvroDeserializeSchema(QuotConfig.config.getProperty("sse.topic")), properties);
        sseKafka.setStartFromEarliest();
        DataStreamSource<SseAvro> sseSource = env.addSource(sseKafka);
//        sseSource.print("沪市：");

        //消费深市行情
        FlinkKafkaConsumer011<SzseAvro> szseKafka = new FlinkKafkaConsumer011<SzseAvro>(QuotConfig.config.getProperty("szse.topic"), new AvroDeserializeSchema(QuotConfig.config.getProperty("szse.topic")), properties);
        szseKafka.setStartFromEarliest();
        DataStreamSource<SzseAvro> szseSource = env.addSource(szseKafka);
//        szseSource.print("深市：");

        // 7.数据过滤（时间和null字段）
        //沪市过滤
        SingleOutputStreamOperator<SseAvro> sseFilter = sseSource.filter(new FilterFunction<SseAvro>() {
            @Override
            public boolean filter(SseAvro value) throws Exception {
                //时间过滤和数据是否为0过滤
                return QuotUtil.checkTime(value) && QuotUtil.checkData(value);
            }
        });

        //深市过滤
        SingleOutputStreamOperator<SzseAvro> szseFilter = szseSource.filter(new FilterFunction<SzseAvro>() {
            @Override
            public boolean filter(SzseAvro value) throws Exception {
                return QuotUtil.checkTime(value) && QuotUtil.checkData(value);
            }
        });

        //8.数据转换、合并
        DataStream<CleanBean> unionData = sseFilter.map(new SseMap()).union(szseFilter.map(new SzseMap()));
//        unionData.print("=====");

        //9.过滤个股数据
        SingleOutputStreamOperator<CleanBean> filterData = unionData.filter(new FilterFunction<CleanBean>() {
            @Override
            public boolean filter(CleanBean value) throws Exception {
                return QuotUtil.isStock(value);
            }
        });
        filterData.print("个股数据：");

        //10.设置水位线
        filterData.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<CleanBean>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(CleanBean element) {
                Long eventTime = element.getEventTime();
                return eventTime;
            }
        });

        SingleOutputStreamOperator<CleanBean> waterData = filterData.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<CleanBean>() {
            //延时时间
            Long delayTime = Long.valueOf(QuotConfig.config.getProperty("delay.time"));
            //当前水位线时间
            Long currentTimestamp = 0L;

            @Nullable
            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(currentTimestamp - delayTime);
            }

            @Override
            public long extractTimestamp(CleanBean element, long previousElementTimestamp) {

                Long eventTime = element.getEventTime();
                //保证整个时间轴一直往前走，谁大取谁
                currentTimestamp = Math.max(eventTime, currentTimestamp);
                return eventTime;
            }
        });
//        waterData.print("水位线：");
        /**
         * 1.个股秒级行情
         * 2.分时行情
         * 3.分时行情备份至HDFS
         * 4.涨跌幅行情
         */
//        //1.个股秒级行情
        new StockSecTask().process(waterData);
//        //2.分时行情
        new StockMinTask().process(waterData);
////        //3.分时行情备份至HDFS
        new StockMinPutHdfsTask().process(waterData);
//        //4.涨跌幅行情
        new StockIncrTask().process(waterData);
        //触发执行
        env.execute("stock stream");
    }

}
